// -*-mode:c++; coding:utf-8-*-

#include "logger.hpp"
#include "naming_strategy.hpp"

#include "location_cache.hpp"

namespace xbase
{
	typedef location_cache::meta_data_map meta_data_map;
	
	bool location_cache::request_item::is_whole_table_scan() const
	{
		const std::string whole_table_start_row = naming_strategy::make_scan_root_meta_start_row(m_table_name, "");
		const std::string whole_table_end_row = naming_strategy::make_scan_root_meta_end_row(m_table_name, "");
		
		return (m_start_row==whole_table_start_row && m_end_row==whole_table_end_row);
	}

	std::string location_cache_key::to_string(const location_cache_key &key)
	{
		std::ostringstream out;
		out << "{tablet_name: " << key.table_name
			<< "; start_row: " << dump_string_hex(key.start_row)
			<< "}";
		return out.str();
	}

	std::string location_cache_key::to_string()
	{
		return to_string(*this);
	}

	std::string location_cache_value::to_string(const location_cache_value &value)
	{
		std::ostringstream out;
		out << "{meta_got_time: " << value.meta_got_time
			<< "; tablet_info: " << value.tablet_info_p->to_string()
			<< "; server_addr: " << value.server_addr_p->to_string()
			<< "}";
		return out.str();
	}

	std::string location_cache_value::to_string()
	{
		return to_string(*this);
	}

	void location_cache::on_async_get_meta(
		handler_type handler,
		std::string table_name,
		std::string request_row,
		bool hard,
		uint64_t old_meta_got_time)
	{
		error_code ec;

		LOG(INFO, "async_get_meta table_name: " << table_name
			<< " request_row: " << dump_string_hex(request_row)
			<< " hard: " << hard
			<< " old_meta_got_time: " << old_meta_got_time
			);

		meta_data_map::iterator iter;
		if (lookup(table_name, request_row, iter))
		{
			location_cache_value meta = iter->second;

			protocol::tablet_info_ptr tablet_info_p = iter->second.tablet_info_p;
			protocol::server_address_ptr server_addr_p = iter->second.server_addr_p;
			uint64_t meta_got_time = iter->second.meta_got_time;

			LOG(TRACE, "meta data is in the cache,"
				<< " tablet_info: " << tablet_info_p->to_string()
				<< " server_address: " << server_addr_p->to_string()
				<< " meta_got_time: " << meta_got_time
				);

			if (!hard)
			{
				LOG(DEBUG, "it's not hard search, return directly ...");

				// 在缓冲找到就不必查找数据库
				dispatch(request_row, handler, meta, ec);
			}
			else
			{
				if (meta_got_time > old_meta_got_time)
				{
					LOG(DEBUG, "info in cache is renewed, return directly ...");

					// 上层使用的元数据是更旧的版本，直接返回当前版本
					dispatch(request_row, handler, meta, ec);
				}
				else
				{
					// 上层使用的元数据是当前版本，需要查询根子表
					std::string scan_start_row = naming_strategy::make_scan_root_meta_start_row(table_name,
						iter->second.tablet_info_p->m_start_row);
					std::string scan_end_row = naming_strategy::make_scan_root_meta_end_row(table_name,
						iter->second.tablet_info_p->m_end_row);

					LOG(INFO, "info in cache is expired, need to reload meta data");

					make_new_request(handler, table_name, request_row, scan_start_row,
						scan_end_row, old_meta_got_time);
				}
			}
		}
		else
		{
			LOG(INFO, "not found in cache, need to load whole table for " << table_name);

			// 需要扫描整张表的信息
			std::string scan_start_row = naming_strategy::make_scan_root_meta_start_row(table_name, "");
			std::string scan_end_row = naming_strategy::make_scan_root_meta_end_row(table_name, "");

			make_new_request(handler, table_name, request_row, scan_start_row,
				scan_end_row, cur_micro_seconds());
		}
	}

	void location_cache::on_got_root_meta(const error_code& ec)
	{
		LOG(INFO, "on_got_root_meta ec: " << ec << " - " << ec.message());

		if (!ec)
		{
			m_root_meta_valid = true;

			process_new_request();
		}
		else
		{
			// 获得根子表位置发生错误，说明master错误，直接向上返回错误码
			while (!m_meta_request_list.empty())
			{
				request_item_ptr item_ptr = m_meta_request_list.front();

				dispatch(item_ptr->m_request_row, item_ptr->m_handler, location_cache_value(), ec);

				m_meta_request_list.pop_front();
			}
		}
	}

	void location_cache::on_got_next_row(meta_row_result::ptr_type meta_row_ptr, const error_code &ec)
	{
		request_item_ptr item_ptr;

		LOG(INFO, "on_got_next_row ec: " << ec << " - " << ec.message());

		assert(!m_meta_request_list.empty());

		if (!ec)
		{
			LOG(INFO, "got row, tablet_info: " << meta_row_ptr->m_tablet_info.to_string()
				<< " server_address: " << meta_row_ptr->m_server_address.to_string()
				);

			item_ptr = m_meta_request_list.front();

			uint16_t tablet_status = meta_row_ptr->m_tablet_info.m_status;

			if (!TabletStatusGetSplitting(tablet_status)
				&& !TabletStatusGetDeleted(tablet_status))
			{
				// 更新cache
				update(meta_row_ptr);
			}

			do_next_meta_row_scan(item_ptr);
		}
		else
		{
			if (ec != xbase::error::eos)
			{
				// 子表不存在或者是子表服务器连不上，需要重查根子表位置
				LOG(INFO, "root meta address is wrong, reload ...");

				// 本次扫描撤销，重置扩展扫描的范围
				item_ptr = m_meta_request_list.front();
				item_ptr->m_range_expand_times = 0;

				m_root_meta_valid = false;

				do_root_meta_query();
			}
			else
			{
				item_ptr = m_meta_request_list.front();
				m_meta_request_list.pop_front();

				LOG(INFO, "meta scan for " << item_ptr->m_table_name
					<< "[" << dump_string_hex(item_ptr->m_request_row) << "] finished,"
					<< " valid_row_count=" << item_ptr->m_meta_tablet_scanner_ptr->get_valid_row_count()
					);

				if (item_ptr->is_whole_table_scan())
				{
					LOG(INFO, "it's whole table request for " << item_ptr->m_table_name);

					if (item_ptr->m_meta_tablet_scanner_ptr->get_valid_row_count() == 0)
					{
						// 在cache中清除该表的所有元数据缓存
						meta_data_map::iterator delete_start_iter, delete_end_iter;
						std::string request_row = "";
						if (lookup(item_ptr->m_table_name, request_row, delete_start_iter))
						{
							delete_end_iter = delete_start_iter;
							++delete_end_iter;
							while (delete_end_iter != m_meta_data_map.end()
								&& delete_end_iter->first.table_name == item_ptr->m_table_name)
							{
								++delete_end_iter;
							}

							m_meta_data_map.erase(delete_start_iter, delete_end_iter);
						}

						dispatch(item_ptr->m_request_row, item_ptr->m_handler,
							location_cache_value(), xbase::error::no_tablet_loaded);
					}
					else
					{
						meta_data_map::iterator iter;
						lookup(item_ptr->m_table_name, item_ptr->m_request_row, iter);

						assert(iter!=m_meta_data_map.end());

						dispatch(item_ptr->m_request_row, item_ptr->m_handler, iter->second, xbase::error::ok);
					}
				}
				else
				{
					meta_data_map::iterator iter;
					lookup(item_ptr->m_table_name, item_ptr->m_request_row, iter);

					if (item_ptr->m_meta_tablet_scanner_ptr->get_valid_row_count() == 0)
					{
						// 一行子表信息都没有返回，说明查询的范围已经是某个合并后子表的真子集
						// 扩展范围后，重新查找

						expand_range(iter, item_ptr);

						LOG(INFO, "request row: " << dump_string_hex(item_ptr->m_request_row)
							<< "range expand to [" << dump_string_hex(item_ptr->m_start_row) << ","
							<< item_ptr->m_end_row << "]");

						m_meta_request_list.push_front(item_ptr);
					}
					else
					{
						dispatch(item_ptr->m_request_row, item_ptr->m_handler, iter->second, xbase::error::ok);
					}
				}

				process_new_request();
			}
		}
	}

	void location_cache::process_new_request()
	{
		if (m_meta_request_list.empty())
		{
			return;
		}

		LOG(INFO, "1) pending new request num: " << m_meta_request_list.size());

		meta_data_map::iterator iter;
		request_item_ptr item_ptr = m_meta_request_list.front();

		// 请求的行有可能已经得到更新
		while (true)
		{
			if (lookup(item_ptr->m_table_name, item_ptr->m_request_row, iter)
				&& item_ptr->m_old_meta_got_time < iter->second.meta_got_time)
			{
				LOG(INFO, "meta for request_row: " << item_ptr->m_request_row << " has been updated");

				// 请求的行所在块已经得到更新，返回给上层
				dispatch(item_ptr->m_request_row, item_ptr->m_handler, iter->second, xbase::error::ok);

				m_meta_request_list.pop_front();

				if (m_meta_request_list.empty())
				{
					break;
				}
				else
				{
					item_ptr = m_meta_request_list.front();
				}
			}
			else
			{
				break;
			}
		}

		if (!m_meta_request_list.empty())
		{
			LOG(INFO, "2) pending new request num: " << m_meta_request_list.size());

			item_ptr = m_meta_request_list.front();
			do_new_meta_scan(item_ptr);
		}
	}

	void location_cache::expand_range(meta_data_map::iterator iter, request_item_ptr item_ptr)
	{
		unsigned steps;
		unsigned max_steps = m_max_update_steps<<(item_ptr->m_range_expand_times);

		std::string table_name = iter->first.table_name;

		LOG(INFO, "expand range, table_name: " << table_name
			<< " request_row: " << item_ptr->m_request_row
			<< " max_steps: " << max_steps);

		meta_data_map::iterator backward_iter = iter;
		steps = 0;
		while (steps < max_steps
			&& backward_iter != m_meta_data_map.begin()
			&& backward_iter->first.table_name == table_name)
		{
			--backward_iter;
			++steps;
		}

		if (backward_iter->first.table_name != table_name)
		{
			++backward_iter;
		}

		meta_data_map::iterator forward_iter = iter;
		steps = 0;
		while (steps < max_steps
			&& forward_iter != m_meta_data_map.end()
			&& forward_iter->first.table_name == table_name)
		{
			++forward_iter;
			++steps;
		}

		if (forward_iter == m_meta_data_map.end()
			|| forward_iter->first.table_name != table_name)
		{
			--forward_iter;				
		}

		item_ptr->m_start_row = naming_strategy::make_scan_root_meta_start_row(table_name, backward_iter->second.tablet_info_p->m_start_row);
		item_ptr->m_end_row = naming_strategy::make_scan_root_meta_end_row(table_name, forward_iter->second.tablet_info_p->m_end_row);

		item_ptr->m_range_expand_times++;
	}

	meta_data_map::iterator location_cache::update(meta_row_result::ptr_type meta_row_ptr)
	{
		std::string &table_name = meta_row_ptr->m_tablet_info.m_table_name;
		std::string &start_row = meta_row_ptr->m_tablet_info.m_start_row;
		std::string &end_row = meta_row_ptr->m_tablet_info.m_end_row;
		meta_data_map::iterator iter;

		LOG(INFO, "update " << table_name << "[" << dump_string_hex(start_row)
			<< " , " << dump_string_hex(end_row) << "]");

		if (!lookup(table_name, start_row, iter))
		{
			LOG(INFO, table_name << " : " << dump_string_hex(start_row) << " doesn't exists");

			location_cache_key key;
			key.table_name = table_name;
			key.start_row = start_row;

			location_cache_value value;
			value.tablet_info_p = protocol::tablet_info_ptr(new protocol::tablet_info(meta_row_ptr->m_tablet_info));
			value.server_addr_p = protocol::server_address_ptr(new protocol::server_address(meta_row_ptr->m_server_address));
			value.meta_got_time = cur_micro_seconds();

			iter = m_meta_data_map.upper_bound(key);
			if (iter != m_meta_data_map.end()
				&& iter->first.table_name == table_name
				&& ((end_row=="") || iter->second.tablet_info_p->m_start_row < end_row))
			{
				meta_data_map::iterator start_iter = iter;
				meta_data_map::iterator end_iter = iter;

				// 找到新段尾部所在的旧段
				while (end_iter != m_meta_data_map.end()
					&& end_iter->second.tablet_info_p->m_table_name == table_name
					&& end_iter->second.tablet_info_p->m_end_row != ""
					&& (end_row == "" || end_iter->second.tablet_info_p->m_end_row < end_row))
				{
					end_iter++;
				}

				if (end_iter != m_meta_data_map.end()
					&& ((end_row!="" && end_iter->second.tablet_info_p->m_end_row=="")
					|| (end_row!="" && end_row < end_iter->second.tablet_info_p->m_end_row)))
				{
					LOG(INFO, "new meta end row: " << dump_string_hex(end_row) << " locates in original range ["
						<< dump_string_hex(end_iter->second.tablet_info_p->m_start_row) << " , "
						<< dump_string_hex(end_iter->second.tablet_info_p->m_end_row) << "]"
						);

					// 旧段从中间分开，向map插入尾部的一段
					// 插入的这一段的信息和原来的旧段一样
					location_cache_key key;
					key.table_name = table_name;
					key.start_row = end_row;

					location_cache_value value;
					value.tablet_info_p = protocol::tablet_info_ptr(new protocol::tablet_info(*(end_iter->second.tablet_info_p)));
					value.tablet_info_p->m_start_row = end_row;
					value.server_addr_p = protocol::server_address_ptr(new protocol::server_address(*(end_iter->second.server_addr_p)));

					end_iter->second.tablet_info_p->m_start_row = end_row;

					LOG(INFO, "need to split old meta");
					LOG(INFO, "new meta A: {key: {" << location_cache_key::to_string(end_iter->first)
						<< "}; value: {" << end_iter->second.to_string() << "}");
					LOG(INFO, "new meta B: {key: {" << location_cache_key::to_string(key)
						<< "}; value: {" << value.to_string() << "}");

					m_meta_data_map.insert(end_iter, meta_data_map::value_type(key, value));
				}

				// 如果新旧两段的末尾都是表尾，则旧段应该删除
				if (end_row==""
					&& end_iter != m_meta_data_map.end()
					&& end_iter->second.tablet_info_p->m_end_row=="")
				{
					end_iter++;
				}

				LOG(INFO, "erase some meta: ");
				LOG(INFO, "from meta: {key: {" << location_cache_key::to_string(start_iter->first) << "};"
					<< " value: {" << start_iter->second.to_string() << "};}"
					);

				if (end_iter!=m_meta_data_map.end())
				{
					LOG(INFO, "to meta: {key: {" << location_cache_key::to_string(end_iter->first) << "};"
						<< " value: {" << end_iter->second.to_string() << "};}"
						);
				}
				else
				{
					LOG(INFO, "to meta: {m_meta_data_map.end()}");
				}

				m_meta_data_map.erase(start_iter, end_iter);
			}

			// 向map中插入新段
			LOG(INFO, "insert new meta:");
			LOG(INFO, "new meta: {key: {" << key.to_string() << "};"
				<< " value: {" << value.to_string() << "};}"
				);

			return m_meta_data_map.insert(meta_data_map::value_type(key, value)).first;
		}
		else
		{
			LOG(INFO, table_name << " : " << dump_string_hex(start_row) << " locates in original range "
				<< iter->first.table_name << "["
				<< dump_string_hex(iter->second.tablet_info_p->m_start_row) << " , "
				<< dump_string_hex(iter->second.tablet_info_p->m_end_row) << "]");

			meta_data_map::iterator start_iter = iter;
			meta_data_map::iterator end_iter;

			// 新段的起始行在某个旧段中间，把旧段的尾部截短
			if (start_row > iter->second.tablet_info_p->m_start_row)
			{
				location_cache_key key;
				key.table_name = table_name;
				key.start_row = start_row;

				location_cache_value value;
				value.tablet_info_p = protocol::tablet_info_ptr(new protocol::tablet_info(*(iter->second.tablet_info_p)));
				value.tablet_info_p->m_start_row = start_row;
				value.server_addr_p = protocol::server_address_ptr(new protocol::server_address(*(iter->second.server_addr_p)));

				// 旧段从尾部截短
				iter->second.tablet_info_p->m_end_row = start_row;

				LOG(INFO, "need to split old meta");
				LOG(INFO, "new meta A: {key: {" << location_cache_key::to_string(iter->first)
					<< "}; value: {" << iter->second.to_string() << "}");
				LOG(INFO, "new meta B: {key: {" << location_cache_key::to_string(key)
					<< "}; value: {" << value.to_string() << "}");

				m_meta_data_map.insert(iter, meta_data_map::value_type(key, value));
				start_iter++;
			}

			// 找到新段尾部所在的旧段
			end_iter = start_iter;
			while (end_iter != m_meta_data_map.end()
				&& end_iter->second.tablet_info_p->m_table_name == table_name
				&& end_iter->second.tablet_info_p->m_end_row != ""
				&& (end_row == "" || end_iter->second.tablet_info_p->m_end_row < end_row))
			{
				end_iter++;
			}

			if (end_iter != m_meta_data_map.end()
				&& ((end_row!="" && end_iter->second.tablet_info_p->m_end_row=="")
				|| (end_row!="" && end_row < end_iter->second.tablet_info_p->m_end_row)))
			{
				LOG(INFO, "new meta end row: " << dump_string_hex(end_row) << " locates in original range ["
					<< dump_string_hex(end_iter->second.tablet_info_p->m_start_row) << " , "
					<< dump_string_hex(end_iter->second.tablet_info_p->m_end_row) << "]"
					);

				// 旧段从中间分开，向map插入尾部的一段
				// 插入的这一段的信息和原来的旧段一样
				location_cache_key key;
				key.table_name = table_name;
				key.start_row = end_row;

				location_cache_value value;
				value.tablet_info_p = protocol::tablet_info_ptr(new protocol::tablet_info(*(end_iter->second.tablet_info_p)));
				value.tablet_info_p->m_start_row = end_row;
				value.server_addr_p = protocol::server_address_ptr(new protocol::server_address(*(end_iter->second.server_addr_p)));

				end_iter->second.tablet_info_p->m_end_row = end_row;

				LOG(INFO, "need to split old meta");
				LOG(INFO, "new meta A: {key: {" << location_cache_key::to_string(end_iter->first)
					<< "}; value: {" << end_iter->second.to_string() << "}");
				LOG(INFO, "new meta B: {key: {" << location_cache_key::to_string(key)
					<< "}; value: {" << value.to_string() << "}");

				m_meta_data_map.insert(end_iter, meta_data_map::value_type(key, value));
			}

			// 如果新旧两段的末尾都是表尾，则旧段应该删除
			if (end_row==""
				&& end_iter != m_meta_data_map.end()
				&& end_iter->second.tablet_info_p->m_end_row=="")
			{
				end_iter++;
			}

			// 把新段跨越的若干旧块删除
			if (start_iter != end_iter)
			{
				LOG(INFO, "erase some meta: ");
				LOG(INFO, "from meta: {key: {" << location_cache_key::to_string(start_iter->first) << "};"
					<< " value: {" << start_iter->second.to_string() << "};}"
					);

				if (end_iter!=m_meta_data_map.end())
				{
					LOG(INFO, "to meta: {key: {" << location_cache_key::to_string(end_iter->first) << "};"
						<< " value: {" << end_iter->second.to_string() << "};}"
						);
				}
				else
				{
					LOG(INFO, "to meta: {m_meta_data_map.end()}");
				}

				m_meta_data_map.erase(start_iter, end_iter);

				// 向map中插入新段
				location_cache_key key;
				key.table_name = table_name;
				key.start_row = start_row;

				location_cache_value value;
				value.tablet_info_p = protocol::tablet_info_ptr(new protocol::tablet_info(meta_row_ptr->m_tablet_info));
				value.server_addr_p = protocol::server_address_ptr(new protocol::server_address(meta_row_ptr->m_server_address));
				value.meta_got_time = cur_micro_seconds();

				LOG(INFO, "insert new meta:");
				LOG(INFO, "new meta: {key: {" << key.to_string() << "};"
					<< " value: {" << value.to_string() << "};}"
					);

				return m_meta_data_map.insert(meta_data_map::value_type(key, value)).first;
			}
			else
			{
				LOG(INFO, "write old meta directly");
				LOG(INFO, "old meta: {key: {" << location_cache_key::to_string(start_iter->first) << "};"
					<< " value: {" << start_iter->second.to_string() << "};}"
					);

				start_iter->second.tablet_info_p =protocol::tablet_info_ptr(new protocol::tablet_info(meta_row_ptr->m_tablet_info));
				start_iter->second.server_addr_p = protocol::server_address_ptr(new protocol::server_address(meta_row_ptr->m_server_address));
				start_iter->second.meta_got_time = cur_micro_seconds();

				LOG(INFO, "new meta: {key: {" << location_cache_key::to_string(start_iter->first) << "};"
					<< " value: {" << start_iter->second.to_string() << "};}"
					);

				return start_iter;
			}
		}
	}

	bool location_cache::lookup(std::string &table_name, std::string &request_row,
		meta_data_map::iterator &iter)
	{
		location_cache_key key;
		key.table_name = table_name;
		key.start_row = request_row;

		iter = m_meta_data_map.upper_bound(key);
		if (iter == m_meta_data_map.begin())
		{
			iter = m_meta_data_map.end();
			return false;
		}

		--iter;

		if (iter->first.table_name!=table_name)
		{
			iter = m_meta_data_map.end();
			return false;
		}

		if (iter->second.tablet_info_p->m_end_row != ""
			&& iter->second.tablet_info_p->m_end_row <= request_row)
		{
			iter = m_meta_data_map.end();
			return false;
		}

		return true;
	}

	void location_cache::do_root_meta_query()
	{
		LOG(INFO, "do_root_meta_query()");
		m_root_meta_cache.async_get_root_meta(
			this->get_lower_level_service().wrap(
			boost::bind(&this_type::on_got_root_meta, this_ptr(), _1)
			),
			(! m_root_meta_valid) // hard == !valid, 根子表位置不合法时要强制更新，否则得到的总是旧的！
			);
	}

	void location_cache::do_new_meta_scan(request_item_ptr item_ptr)
	{
		LOG(INFO, "do_new_meta_scan, scan_range: ["
			<< dump_string_hex(item_ptr->m_start_row) << ", "
			<< dump_string_hex(item_ptr->m_end_row) << "]"
			);

		const protocol::server_address &sa = m_root_meta_cache.get_root_server_address();
		endpoint_type ep = sa.to_endpoint();

		item_ptr->m_meta_tablet_scanner_ptr = new_async_module(
			meta_tablet_scanner, m_io_service_pool, ep, m_session_manager,
			item_ptr->m_start_row, item_ptr->m_end_row
			);

		do_next_meta_row_scan(item_ptr);
	}

	void location_cache::do_next_meta_row_scan(request_item_ptr item_ptr)
	{
		item_ptr->m_meta_tablet_scanner_ptr->async_get_next_row(
			this->get_lower_level_service().wrap(
			boost::bind(&this_type::on_got_next_row, this_ptr(), _1, _2)
			)
			);
	}

	void location_cache::make_new_request(handler_type handler, std::string &table_name, std::string &request_row,
		std::string &scan_start_row, std::string &scan_end_row,
		uint64_t old_meta_got_time)
	{
		request_item_ptr item_ptr = request_item_ptr(
			new request_item(handler, table_name, request_row, scan_start_row,
			scan_end_row, old_meta_got_time)
			);

		bool is_requesting = (!m_meta_request_list.empty());
		m_meta_request_list.push_back(item_ptr);

		if (!is_requesting)
		{
			if (m_root_meta_valid)
			{
				LOG(INFO, "root meta is valid, make a meta scan");
				do_new_meta_scan(item_ptr);
			}
			else
			{
				LOG(INFO, "root meta address not valid, reload ...");
				do_root_meta_query();
			}
		}
		else
		{
			LOG(INFO, "there's other request, put into request list");
		}
	}

	void location_cache::dispatch(std::string &request_row, handler_type handler,
		location_cache_value meta, error_code ec)
	{
		LOG(DEBUG, "dispatch for " << dump_string_hex(request_row));

		if (!ec && !TabletStatusGetOnline(meta.tablet_info_p->m_status))
		{
			// 不在线
			ec = xbase::error::tablet_not_exists;
		}

		this->get_upper_level_service().dispatch(
			boost::bind(handler, meta, ec)
			);
	}
}
